package com.llx.dataset

import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.scala.{ExecutionEnvironment, _}
import org.apache.flink.configuration.Configuration


/**
  * val env = ExecutionEnvironment.getExecutionEnvironment
  * val ds1 = env.fromElements("1", "2", "3", "4", "5")
  * val ds2 = env.fromElements("a", "b", "c", "d", "e")
  *
  * ds1.map{
  * t =>
  * (t, ds2)
  * }.print()
  * 运行上述代码会报InvalidProgramException的错，因为在ds1的map算子中无法再去调用ds2，
  * 此时可以使用广播变量将ds2这个变量进行广播，使得ds2这一被广播的数据集在ds1中map算子的所有并行实例中都可用，
  */
object Broadcast广播变量 {
  def main(args: Array[String]): Unit = {
    val env = ExecutionEnvironment.getExecutionEnvironment
    val ds1 = env.fromElements("1", "2", "3", "4", "5")
    val ds2 = env.fromElements("a", "b", "c", "d", "e")

    /**
      * 通过继承map操作的RichMapFunction，或着先关操作的抽象类并实现
      * open与相关操作eg:map
      *在open中 获取广播变量  ds2 = getRuntimeContext.getBroadcastVariable[String]("broadCast").asScala
      *
      * 在外部注册变量.withBroadcastSet(ds2, "broadCast")
      */
    ds1.map(new RichMapFunction[String, (String, String)] {
      private var ds2: Traversable[String] = null

      override def open(parameters: Configuration) {
        import scala.collection.JavaConverters._
        //asScala需要使用隐式转换
        //
        /** 二、获取广播变量
          * getBroadcastVariable 来访问广播变量
          */
        ds2 = getRuntimeContext.getBroadcastVariable[String]("broadCast").asScala
      }

      def map(t: String): (String, String) = {
        var result = ""
        for (broadVariable <- ds2) {
          result = result + broadVariable + " "
        }
        (t, result)
      }

      /** 一、注册广播变量
        * withBroadcastSet 来注册  广播变量
        */
    }).withBroadcastSet(ds2, "broadCast").print()
  }

}
